package com.kd.crawler.main;

import java.util.LinkedList;
import java.util.Properties;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.jms.Message;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.kd.crawler.common.jms.ActiveMQConsumer;
import com.kd.crawler.common.jms.ActiveMQConsumer.ReceiveListener;
import com.kd.crawler.common.utils.PropertiesUtils;
import com.kd.crawler.entity.CrawlerEntry;
import com.kd.crawler.parser.thread.MQMessageSender;
import com.kd.crawler.parser.thread.PageParserListener;
import com.kd.crawler.parser.thread.PageParserThread;
import com.kd.crawler.parser.thread.URLFilterQueue;

public class ParserMain {	
	
	private Logger logger = Logger.getLogger(ParserMain.class);

	private Queue<PageParserThread> spiderQueue = new LinkedList<PageParserThread>();
	
	public void start() {
		new Thread(new URLFilterQueue()).start();
	
		new Thread(new MQMessageSender()).start();
		
		Properties proMQ = PropertiesUtils.getProperties("activeMQ.properties");
		Properties ProParser = PropertiesUtils.getProperties("parse.properties");
		String threadCount = ProParser.getProperty("thread_count");
		int defaultThreadCount = 1;
		if (threadCount != null) {
			defaultThreadCount = Integer.valueOf(threadCount);
		}
		final ExecutorService pool = Executors.newFixedThreadPool(defaultThreadCount);
		for (int i = 0; i < defaultThreadCount; i++) {
			offerSpider(new PageParserThread(new PageParserListener<PageParserThread>() {
				public void onFinish(PageParserThread source) {
					offerSpider(source);
					
				}
				
			}));
		}	
		System.out.println("正在接收信息");
		while (true) {
			try {

				ActiveMQConsumer.recevieQueueMessage(proMQ.getProperty("QUEUE_CRAWL_HTML_MESSAGE_OLD"), new ReceiveListener() {

				ActiveMQConsumer.recevieQueueMessage(proMQ.getProperty("QUEUE_CRAWL_HTML_MESSAGE"), new ReceiveListener() {

					public void onReceive(Message message) {
						if (message == null) {
							return;
						}
						String messageText;
						try {
							messageText = ((TextMessage) message).getText();
							logger.debug("从队列接收到入口信息:" + messageText);
							PageParserThread spider = pollSpider();
							spider.setCrawlerEntry(JSON.parseObject(messageText, CrawlerEntry.class));
							pool.execute(spider);
							message.acknowledge();
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				});
			} catch (Exception e) {
				logger.error("接收activeMessage报错:", e);
			}
		}
	}
	private synchronized PageParserThread pollSpider() {
		PageParserThread spider = null;
		while (spider == null) {
			spider = spiderQueue.poll();
			if (spider == null) {
				try {
					this.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		return spider;
	}
	private synchronized void offerSpider(PageParserThread spider) {
		spiderQueue.offer(spider);
		this.notify();
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		new ParserMain().start();
	}
	
}

